package com.rxjava.simpleio3;

import com.rxjava.myrxjava.os.Handler;

/**
 * 被观察者，用来生产发送事件；
 */
public class Observable<T> {

    private OnSubscribe<T> onSubscribe;

    private Observable(OnSubscribe<T> onSubscribe) {
    	//onSubscribe -> subscriber.onNext(10);
    	//onSubscribe -> OnSubscribeLift对象(OperatorMap对象(return "map" + integer;))
    	//onSubscribe -> OnSubscribeLift对象(OperatorObserveOn对象)
    	//onSubscribe -> OnSubscribeLift对象(OperatorMap对象(return new User(name);))
    	//onSubscribe -> OnSubscribeOnIO对象
        this.onSubscribe = onSubscribe;
    }

    public final void subscribe(Subscriber<? super T> subscriber) {
    	System.out.println("step1");
    	//subscriber -> System.out.println(u);
    	//onSubscribe -> OnSubscribeOnIO对象
    	//subscriber -> System.out.println(u);
    	//onSubscribe -> OnSubscribeLift对象(OperatorMap对象(return new User(name);))
    	onSubscribe.call(subscriber);
    }

    public static <T> Observable<T> create(OnSubscribe<T> onSubscribe) {
        return new Observable<>(onSubscribe);
    }

    public final <R> Observable<R> map(Func1<? super T, ? extends R> func) {
    	//1map -> func -> return "map" + integer;
    	//2map -> func -> return new User(name);
        return lift(new OperatorMap<T, R>(func));
    }

    public final <R> Observable<R> lift(final Operator<? extends R, ? super T> operator) {
    	//operator -> OperatorMap对象(return "map" + integer;)
    	//operator -> OperatorObserveOn对象
    	//operator -> OperatorMap对象(return new User(name);)
        return new Observable<>(new OnSubscribeLift<T, R>(onSubscribe, operator));
    }

    public final Observable<T> subscribeOnIO() {
    	//this -> Observable对象(OnSubscribeLift对象(OperatorMap对象(return new User(name);)))
        return create(new OnSubscribeOnIO<T>(this));
    }

    public final Observable<T> observeOn(Handler handler) {
        return lift(new OperatorObserveOn<T>(handler));
    }

}